package producerConsumer;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Semaphore;

/**
 * @Author: linKe
 * @Date: 2022/1/15 14:29
 * @FileName: SemaphoreBasedChannel
 * @version: 1.0
 * @Description: 使用Semaphore控制生产者的速度--处理通道积压
 * 基于信号量的通道
 * 生产者 消费者模式
 */
public class SemaphoreBasedChannel<E> implements Channel<E> {
    /**
     * 信号量对象
     */
    private final Semaphore semaphore;
    /**
     * 阻塞队列
     */
    private final BlockingQueue<E> queue;

    public SemaphoreBasedChannel(int limit, BlockingQueue<E> queue) {
        super();
        this.semaphore = new Semaphore(limit);
        this.queue = queue;
    }

    /**
     * 加入通道
     *
     * @param e 泛型对象
     * @throws InterruptedException 中断异常
     */
    @Override
    public void put(E e) throws InterruptedException {
        /*
        从这个信号量获取一个许可，阻塞直到有一个可用，或者线程被中断。
            获得一个许可证，如果一个可用并立即返回，将可用许可证的数量减少一个。
            如果没有可用的许可，则当前线程出于线程调度目的而被禁用并处于休眠状态，直到发生以下两种情况之一：
            其他一些线程为此信号量调用release方法，并且当前线程接下来被分配一个许可； 或者
            其他一些线程中断当前线程。
            如果当前线程：
            在进入此方法时设置其中断状态； 或者
            在等待许可时被打断，
            然后抛出InterruptedException并清除当前线程的中断状态。
         */
        semaphore.acquire();
        try {
            /*
            队列添加
             */
            queue.put(e);
        } finally {
            // 释放许可证，将其返回给信号量。
            semaphore.release();
        }
    }

    /**
     * 取出
     *
     * @return 结果
     * @throws InterruptedException 中断异常
     */
    @Override
    public E take() throws InterruptedException {
        // 检索并删除此队列的头部，如有必要，等待元素可用。
        return queue.take();
    }

}
